package com.gmail.at.vssabinin.train_transport_system_part1;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.BasicConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import model.impl.Passenger;

/**
 * 
 * 	Root class of the Server. It's is capable of transmitting and receiving data 
 * over the network. Use console to manage it. It has 3 commands:
 * 	  1. start - starts the Server on the port 1015 and address - "localhost"
 * 	  2. stop - stop server and drop all connection's
 *    3. exit - if server is stopped, stop's main method, and exit this program.
 *    
 * You can use start/stop more than one time.
 * 
 */
public class Server implements Runnable {
	
	//	port to bind the server
	private int port = 0;
	
	// address to bind the server
	private InetAddress serverAddress = null;
	
	private Selector selector = null;
	
	private ServerSocketChannel serverChannel = null;
	
	// show's whether the server stopped or not
	private AtomicBoolean stopped = new AtomicBoolean(false);
	
	// buffer for writing and reading data
	private ByteBuffer buffer = ByteBuffer.allocate(9999);
	
	// client-server protocol for messages
	private Protocol protocol = new Protocol();
	
	// logger for Server
	private Logger logger = LoggerFactory.getLogger("Server");
	
	// the map of socket on which our client (passenger) is bind
	public Map<SocketChannel, Passenger> authorizedUsers = new HashMap<SocketChannel, Passenger>();
	
	// thread pool
	private ExecutorService threadPool = null;
	
	// Map socketChannel shows on which channel it's required to send data from List of ByteBuffer's
	private Map<SocketChannel, List<ByteBuffer>> data = new HashMap<SocketChannel, List<ByteBuffer>>();

	// list of channel's where we should send some data
	private List<SocketChannel> request = new LinkedList<SocketChannel>();
	
	// Task's from read method
	private Map<SocketChannel, byte[]> needToRead = new HashMap<SocketChannel, byte[]>();
		
	
	/**
	 * Constructor for Server.class
	 * 
	 * @param address - address to bind Server
	 * @param bindPort - port on which bind Server
	 * @param threadPool - threadPool where to send tasks
	 */
	public Server(InetAddress address, int bindPort, ExecutorService threadPool) {
		this.port = bindPort;
		this.serverAddress = address;
		this.selector = initSelector();
		this.threadPool =  threadPool;
		this.logger.info("Creating new server on adress: " + address.toString() + " and port " + port);
	}
	
	
	/**
	 * Method shows is Server running now
	 * 
	 * @return whether the server stopped or not
	 */
	public boolean isStopped() {
		
		return stopped.get();
	}
	
	
	/**
	 * Check is it possible to start the given server and if it is -starts it
	 * 
	 * @param server - Server to start 
	 * @return true - if the server was started, false in other cases
	 */
	public boolean turnOnServer (Server server) {
		try {
			// Check, was the initializing OK
			if (selector == null) {
				this.logger.error("Impossible to start server, check given port and address, " +
						"they may be already binded ");
				return false;
			}
			new Thread(server).start();
			this.logger.info("Starting the server");
			return true;
		} catch (Exception e) {
			this.logger.error("Server's launch - failed!!!");
		}
		
		return false;
	}
	
	/**
	 * Method stop the given Server, drop all connections and close server socket.
	 * It checks the variable stopped and using CAS operation sets it to false, so
	 * that - run method of the server in while cycle stops iteration, and server's
	 * thread finishes. Also it check's and close each connection with clients.
	 * 
	 * @param server - indicates which server to stop
	 */
	@SuppressWarnings("finally")
	public void turnOffServer (Server server) {	
		while (!this.stopped.compareAndSet(false, true));
		while (true) {
			if (this.stopped.get() == true) {
				try {
					this.serverChannel.close();
					this.selector.close();
					this.threadPool.shutdown();
					for (SocketChannel s : request) {
						s.close();
					}
				} finally {
					break;
				}
			}
		}
	}
	
	/**
	 * Sends given data to the socketChannel of user
	 * 
	 * @param socketChannel - a channel where to send data, using {@link Protocol}
	 * @param sendingData sending data to socket channel
	 */
	public void send(SocketChannel socketChannel, byte[] sendingData) {
		synchronized  (this.request) {
			this.logger.info("Adding new request to be send");
			// add to request this channel as it's required to send him some data
			this.request.add(socketChannel);
			synchronized (this.data) {
				// check whether there's some data already waiting to be send to the server, if there's some 
				// put this one to the end of queue, if it's empty create new List and put it in the Map data.
				List<ByteBuffer> sendingQueue = (List<ByteBuffer>) this.data.get(socketChannel);
				if (sendingQueue == null) {
					sendingQueue = new ArrayList<ByteBuffer>();
					sendingQueue.add(ByteBuffer.wrap(sendingData));
					this.data.put(socketChannel, sendingQueue);
				} else {
					sendingQueue.add(ByteBuffer.wrap(sendingData));
					this.data.put(socketChannel, sendingQueue);
				}
			}
			// wake up selector - it's required to write some data 
			this.selector.wakeup();
		}
	}
	
	
	public void run() {
		// do while the server is not stopped
		while (!this.stopped.get()) {
			// check is there any data to be write
			synchronized (this.request) {
				Iterator<SocketChannel> changes = this.request.iterator();
				while (changes.hasNext()) {
					SocketChannel change = (SocketChannel) changes.next();
					// check that socketChannel it still connected
					if (change.isConnected()) {
						// there's some data to write to the socket, so make the key WRITE
						change.keyFor(this.selector).interestOps(SelectionKey.OP_WRITE);
					} else {
						// if the socket is not connected - close it
						try {
							this.logger.info("Close disconnected socket");
							change.close();
							this.authorizedUsers.remove(change);
						} catch (IOException e) {
							this.logger.info("Some problems happend during socket close");
						}			
					}
				}
				// all the data waiting to be write to socket channel marked as OP_WRITE, so clear the list 
				// of request
				this.request.clear();
			}
			// switch on selector
			try {
				this.selector.select();
			} catch (Exception e) {
				this.logger.info("Problems with starting selector");
			}
			if (selector.isOpen()) {
				Iterator<SelectionKey> selectedKeys = this.selector.selectedKeys().iterator();
				// check all new events 
				while (selectedKeys.hasNext()) {
					// get new event key and remove it from the set
					SelectionKey newKey = (SelectionKey) selectedKeys.next();
					selectedKeys.remove();
					
					// check that the key is valid
					if (!newKey.isValid()) {
						continue;
					} 
					
					// if it's the event of new connection call the accept on the key
					if (newKey.isAcceptable()) {
						this.accept(newKey);
						
					/* if it's the event of sending some data to the server check the size of given data
					 * using Protocol class if the data is has the expected size, give a new task to thread
					 * pool of workers, and put it to the task queue, if it's bigger take the expected size
					 * and rest live for the next iteration of reading, if it's less than expected size, do
					 * nothing.
					 */
					} else if (newKey.isReadable()) {
						this.read(newKey);
						byte[] bytes = this.needToRead.get((SocketChannel) newKey.channel());
						if (bytes == null) {
							continue;
						}
						ByteBuffer temp = ByteBuffer.wrap(bytes);
						int expectedSize = protocol.decodeMessageLength(temp);
						int actualSize = temp.remaining();
						if (actualSize > expectedSize) {
							byte[] newData = new byte[expectedSize];
							temp.get(newData, 0, expectedSize);	
							this.threadPool.submit(new Worker(new ReceivedData(this, 
									(SocketChannel) newKey.channel(), ByteBuffer.wrap(newData))));
							this.logger.info("Reading new amount of data from client");
						} else if (actualSize == expectedSize) {
							this.threadPool.submit(new Worker(new ReceivedData(this, 
									(SocketChannel) newKey.channel(), temp)));
							this.logger.info("Reading new amount of data from client");
						}
					}
					// if the event key is WRITE call write on this key
					else if (newKey.isWritable()) {
						this.write(newKey);
					}
				}
			}	
		}
		// now tell turnOffServer method, that the loop is finished
		this.stopped.compareAndSet(false, true);
		this.logger.info("Server was stopped!");
	}
	
	
	// initiating selector on this server channel and open the server socket and set configurations
	private Selector initSelector() {
		Selector selectorSocket = null;
		try {
			selectorSocket = Selector.open();
			this.serverChannel = ServerSocketChannel.open();
			// set not blocking mode
			this.serverChannel.configureBlocking(false);
			// bind Server Socket on the given adress
			InetSocketAddress socketAddress = new InetSocketAddress(this.serverAddress, this.port);
			this.serverChannel.socket().bind(socketAddress);
			this.serverChannel.register(selectorSocket, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			if (serverChannel != null) {
				try {
					serverChannel.close();
				} catch (IOException e1) {
					logger.error("Error during close of server channel in initSelector " + 
							e1.getLocalizedMessage());
				}
			}
			if (selectorSocket != null) {
				try {
					selectorSocket.close();
				} catch (IOException e1) {
					logger.error("Error during close of selector socket in initSelector " 
							+ e1.getLocalizedMessage());
				}
			}
			logger.error("Error in initSelector: may be given port already binded!");
			return null;
		}
		return selectorSocket;
	}
	
	// accept new connection from the client
	private void accept(SelectionKey newKey) {
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) newKey.channel();
		SocketChannel socketChannel = null;
		try {
			socketChannel = serverSocketChannel.accept();
			socketChannel.configureBlocking(false);
			socketChannel.register(selector, SelectionKey.OP_READ);
			logger.info("New incoming connection");
		} catch (IOException e) {
			newKey.cancel();
			if (socketChannel != null) {
				try {
					socketChannel.close();
					this.authorizedUsers.remove(socketChannel);
				} catch (IOException e1) {
				}
			}
			logger.error("Error during accepting connection from client " 
					+ e.getLocalizedMessage());
		}
	}
	
	// read new data from given key
	private void read(SelectionKey newKey) {
		// get channel from which data was send and clear buffer
		SocketChannel socketChannel = (SocketChannel) newKey.channel();
		this.buffer.clear();
		int number;
		// check the number of received bytes
		try {
			number = socketChannel.read(this.buffer);
			buffer.flip();
		} catch (IOException e) {
			logger.error("Error during reading data from client " 
					+ e.getLocalizedMessage());
			newKey.cancel();
			// so close the channel and cancel key
			try {
				this.authorizedUsers.remove(socketChannel);
				socketChannel.close();
			} catch (IOException e1) {
				logger.error("Error during closing the channel in read method " 
						+ e1.getLocalizedMessage());
			}
			newKey.cancel();
			
			return;
		}
		
		// the client closed connection so do the same 
		if (number == -1) {
			try {
				newKey.channel().close();
				this.authorizedUsers.remove(newKey.channel());
				logger.info("Closing the channel for client");
			} catch (IOException e) {
				logger.error("Error during closing the channel in read method " 
						+ e.getLocalizedMessage());
			}
			newKey.cancel();
			
			return;
		}
		
		/*
		 * add a new task to workers pool if there is some task's get it and put new one to 
		 * the end of previous;
		 */
		if (this.needToRead.get(socketChannel) == null) {
			this.needToRead.put(socketChannel, this.buffer.array());
		} else {
			byte[] tempByte = new byte[this.buffer.array().length
			                           + this.needToRead.get(socketChannel).length];
			for(int i = 0; i < this.needToRead.get(socketChannel).length; i++) {
				tempByte[i] = this.needToRead.get(socketChannel)[i];
			}
			for(int i = 0; i < this.buffer.array().length; i++) {
				tempByte[i + this.needToRead.get(socketChannel).length] = this.buffer.array()[i];
			}
		}
				
	}

	// write new data to the client 
	private void write(SelectionKey newKey) {
		SocketChannel socketChannel = (SocketChannel) newKey.channel();
		synchronized (this.data) {
			// check whether there's some data waiting to be send, if there is some put the
			// new portion to end of previous
			List<ByteBuffer> sendingQueue = this.data.get(socketChannel);
			while (!sendingQueue.isEmpty()) {
				ByteBuffer temp = sendingQueue.get(0);
				try {
					socketChannel.write(temp);
				} catch (IOException e) {
					newKey.cancel();
					try {
						this.authorizedUsers.remove(socketChannel);
						socketChannel.close();
					} catch (IOException e1) {
						logger.error("Error during closing the socket channel in write method" 
								+ e1.getLocalizedMessage());
					}
					logger.info("Error during writing to the client" + e.getLocalizedMessage());
				}
				// check that the buffer is not overflow
				if (temp.remaining() > 0) {
					break;
				}
				// remove this task
				sendingQueue.remove(0);
			}
			// this key is not waiting for writing some data, make it waiting for READ event
			if (sendingQueue.isEmpty()) {
				newKey.interestOps(SelectionKey.OP_READ);
			}
		}
		
	}
	
	public static void main(String[] args) throws UnknownHostException {
		BasicConfigurator.configure();
		Server server = null;
		Scanner input = new Scanner(System.in);
		while (input.hasNextLine()) {
			String mes = input.next();
			if (mes.equals("start")) {
				server = new Server(InetAddress.getByName("localhost"), 1015, 
							Executors.newCachedThreadPool());
				server.turnOnServer(server);
			}
			if (mes.equals("stop")) {
				server.turnOffServer(server);
			}
			if (mes.equals("exit")) {
				break;
			}
		}
	}
}
